package com.lx.boot.es;

import com.lx.annotation.Note;
import com.lx.boot.web.valid.Validate;
import com.lx.util.LX;
import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
import org.elasticsearch.client.*;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.stream.Collectors;

public class ElasticLowerClient {
    private  org.slf4j.Logger logger= LoggerFactory.getLogger(ElasticLowerClient.class);
    private RestClient client;

    private final static Map<String, ElasticLowerClient> MAPS = new HashMap();

    public static ElasticLowerClient getInstance(EsInfo esInfo) {
        Validate.validate(esInfo);
        String key = esInfo.getHosts()+esInfo.getUsername()+esInfo.getPassword();
        if (!MAPS.containsKey(key)) {
            synchronized (ElasticLowerClient.class) {
                if (!MAPS.containsKey(key)) {
                    MAPS.put(key, new ElasticLowerClient(esInfo));
                }
            }
        }
        return MAPS.get(key);
    }
    ElasticLowerClient(EsInfo esInfo) {
        final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
        credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(esInfo.getUsername(), esInfo.getPassword()));
        String[] hostsAndPorts=esInfo.getHosts().split(",");
        HttpHost[] httpHosts = new HttpHost[hostsAndPorts.length];
        for(int i=0;i<hostsAndPorts.length;i++){
            httpHosts[i] = HttpHost.create(hostsAndPorts[i]);
        }
        client = RestClient.builder(httpHosts).setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
            @Override
            public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {
                httpClientBuilder.disableAuthCaching();
                return httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
            }
        }).build();
    }

    public void insertList(List<EsData> list, String baseIndex) throws IOException {
        StringBuffer sendStr=new StringBuffer();
        for(int a=0;a<list.size();a++) {
            EsData map=list.get(a);
            String ent="{\"index\":{\"_id\":\""+ System.currentTimeMillis()+LX.uuid32(10)+"\"}} ";
            sendStr.append(ent);
            sendStr.append("\r\n");
            sendStr.append(LX.toJSONString(map.getData()));
            sendStr.append("\r\n");
        }
        String endpoint = "/"+baseIndex+"/_bulk";
        Request request = new Request("POST", endpoint);
        request.setJsonEntity(sendStr.toString());
        client.performRequestAsync(request, new ResponseListener() {
            @Override
            public void onSuccess(Response response) {
            }
            @Override
            public void onFailure(Exception e) {
                logger.error("ElasticSearch commit Failure!",e);
            }
        });
    }

    //说明: 查询数据
    /**{ ylx } 2021/1/14 13:30 */
    public String query(String queryStr, String index){
        String[] indexs=index.split(",");
        List<String> reindexs= getExistIndices(indexs);
        String indexStr=String.join(",",reindexs);
        if("".equals(indexStr)){
            return "";
        }
        String reStr="";
        Request request = new Request("POST", indexStr+"/_search");
        request.setJsonEntity(queryStr);
        try {
            Response res=client.performRequest(request);

            InputStream inputStream=res.getEntity().getContent();
            byte[] bytes = new byte[0];
            bytes = new byte[inputStream.available()];
            inputStream.read(bytes);
            String str = new String(bytes);
            reStr=str;
        } catch (Exception e) {
            reStr = "";
        }
        return reStr;
    }

    //查询index列表
    public String cat(String index){
        String reStr="";
        Request request = new Request("GET", "/_cat/indices/"+index+"?v");
        try {
            Response res=client.performRequest(request);

            InputStream inputStream=res.getEntity().getContent();
            byte[] bytes = new byte[0];
            bytes = new byte[inputStream.available()];
            inputStream.read(bytes);
            String str = new String(bytes);
            reStr=str;
        } catch (Exception e) {
            reStr = "";
        }
        return reStr;
    }
    public boolean deleteIndex(String index){
        try {
            Request request = new Request("DELETE", "/" + index + "");
            Response res = client.performRequest(request);
            if (res.getStatusLine().getStatusCode() == 200) {
                return true;
            }
        }catch (Exception e){
            return false;
        }
        return false;
    }
    public void close(){
        try {
            client.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public List<String> getExistIndices(String [] indices){
        List<String> existIndexList = new ArrayList<String>();
        for (String index: indices){
            try {
                Request request = new Request("HEAD", "/"+index+"");
                Response res=client.performRequest(request);
                if(res.getStatusLine().getStatusCode()==200){
                    existIndexList.add(index);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        return existIndexList;
    }





}
